package com.shujia.flink.sink

import java.io.{BufferedWriter, FileWriter}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object Demo2SinkFile {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    val ds = env.socketTextStream("node1", 8888)

    ds.addSink(new FileSink("flink/data/sink"))

    env.execute("sink")
  }

}

/**
  * 自定义sink
  *
  */
class FileSink(path: String) extends RichSinkFunction[String] {

  val _path: String = path

  var buffer: BufferedWriter = _

  /**
    * open close  方法  在每一个task中执行一次
    *
    */
  override def open(parameters: Configuration): Unit = {
    val fileWriter = new FileWriter(_path)
    buffer = new BufferedWriter(fileWriter)
  }


  override def close(): Unit = {
    //关闭连接
    buffer.close()
  }

  //每一个行数据咨询一次
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
    buffer.write(value)
    buffer.newLine()
    buffer.flush()
  }

}
